	package com.graph6;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class GroupDemoThredPool {
	private static final Logger log = LoggerFactory.getLogger(GroupDemoThredPool.class);

	public static void main(String[] args) {
		Random r = new Random();
		int[][] edges = new int[10000][2];
		
		for (int[] edge : edges) {
			edge[0]=r.nextInt(9999);
			edge[1]=r.nextInt(9999);
			
		}
		//排序
		Arrays.sort(edges, new Comparator<int[]>() {
		    @Override
		    public int compare(int[] o1, int[] o2) {
		        return o1[0] - o2[0];
		    }
		});
		
		List<Tx> txs = new ArrayList<Tx>();
		for (int[] edge : edges) {
			txs.add(new Tx(edge[0], edge[1]));
		}
		HashMap<Integer, List<Tx>> map = new HashMap<Integer, List<Tx>>();

		int group = 0;
		while (!txs.isEmpty()) {
			Tx vertex = txs.get(0);
			Set<Integer> set = new HashSet<Integer>();
			set.add(vertex.getFrom());
			set.add(vertex.getTo());
			Iterator<Tx> iterator = txs.iterator();
			List<Tx> txadd = new ArrayList<Tx>();
			while (iterator.hasNext()) {
				Tx tx = iterator.next();
				if(set.contains(tx.getFrom())||set.contains(tx.getTo())) {
					set.add(tx.getFrom());
					set.add(tx.getTo());
					txadd.add(tx);
					iterator.remove();
				}
			}
			map.put(group, txadd);
			group++;
		}
		System.out.println("总共分组："+map.size());

		// 获取服务器 CPU 数量
		int cpus = Runtime.getRuntime().availableProcessors();
		// 初始化线程池
		ExecutorService threadPool  = new ThreadPoolExecutor(
				cpus,//核心池线程数大小
				cpus*1000,//最大线程数大小
				3L,//超时间等待时间
				TimeUnit.SECONDS,//时间单位
				new LinkedBlockingDeque<>(3),//阻塞队列
				Executors.defaultThreadFactory(),//线程工厂
				new ThreadPoolExecutor.AbortPolicy() //拒绝策略
		);
		/**
		 * 1、ThreadPoolExecutor.AbortPolicy(); 满了就会抛出异常，丢弃任务
		 * 2、ThreadPoolExecutor.CallerRunsPolicy(); 哪来的去哪里，找对应的线程执行
		 * 3、ThreadPoolExecutor.DiscardOldestPolicy(); 尝试获取任务，不一定执行
		 * 4、ThreadPoolExecutor.DiscardPolicy(); 不抛出异常，丢弃任务
		 */
		LinkedBlockingQueue<Tx> vectors = new LinkedBlockingQueue<Tx>();
		final CountDownLatch countDownLatch = new CountDownLatch(10000);
		 
		//TODO 根据时间排序
		
		//线程池的使用方式
		try {
			// 遍历所有分组
			map.forEach((key, value) -> {
				
				threadPool.execute(()->{
					
					// 执行每个分组
					for (Tx tx : value) {
						
						try {
							TimeUnit.MILLISECONDS.sleep(20);
							vectors.add(tx);
							countDownLatch.countDown();
							//log.info("group {}, tx: {}",key , tx);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}

				});

			});
			System.out.println("主线程等待工作线程执行");
	        countDownLatch.await();
	        for (Tx tx: vectors) {
	            System.out.println(tx);
	        }
			
			
			/*
			while(!map.isEmpty()) {
				int count =0;
				for(Iterator<Entry<Integer, List<Tx>>> iterator = map.entrySet().iterator();iterator.hasNext();){
					if(count < cpus) {
						count++;
						Entry<Integer, List<Tx>> keyObject = iterator.next();
						
						threadPool.execute(() -> {
							for(Tx tx : keyObject.getValue()) {
								try {
									TimeUnit.MILLISECONDS.sleep(20);
									log.info("tx: {}", tx);
								} catch (Exception e) {
									e.printStackTrace();
								}
							}
			                iterator.remove();
						});
					}else {
						 break;
					}
	            }
			}
			*/
			/*
			while(!map.isEmpty()) {
				int count =0;
				for(Iterator<Entry<Integer, List<Tx>>> iterator = map.entrySet().iterator();iterator.hasNext();){
					if(count < cpus) {
						count++;
						Entry<Integer, List<Tx>> keyObject = iterator.next();
						
						for(Tx tx : keyObject.getValue()) {
							log.info("tx: {}", tx);
						}
						
						iterator.remove();
					}else {
						break;
					}
				}
			}*/
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			System.out.println("size:"+vectors.size());
			// 线程池使用完后需要关闭
			threadPool.shutdown();
		}
	}
}
